package com.shujia.flink.core;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;

public class Demo8CoMap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> wordsDS = env.socketTextStream("master", 8888);
        DataStream<String> linesDS = env.socketTextStream("master", 9999);

        //connect --> CoFlatMap 相当于FlatMap --> union
        ConnectedStreams<String, String> connectDS = wordsDS.connect(linesDS);
        DataStream<Tuple2<String, Integer>> coDS = connectDS
                .flatMap(new CoFlatMapFunction<String, String, Tuple2<String, Integer>>() {
                    @Override
                    public void flatMap1(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                        out.collect(Tuple2.of(value, 1));
                    }

                    @Override
                    public void flatMap2(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                        for (String word : value.split(",")) {
                            out.collect(Tuple2.of(word, 1));
                        }
                    }
                });

        coDS.print();

        env.execute();
    }
}